package com.gzhryc.socket.mosquitto;

import com.gzhryc.common.DateTools;
import com.gzhryc.common.logging.Logger;
import com.gzhryc.socket.SocketCallback;
import com.gzhryc.socket.SocketEvent;
import com.gzhryc.socket.SocketMessage;
import com.gzhryc.socket.SocketNode;
import com.gzhryc.socket.message.MqttSocketMessage;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;

import java.util.Date;
import java.util.concurrent.atomic.AtomicBoolean;

public class MosquittoClientMqttCallback implements org.eclipse.paho.client.mqttv3.MqttCallback{

	static Logger log = Logger.getLogger(MosquittoClientMqttCallback.class);
	
	final SocketNode node;
	final SocketCallback callback;
	
	final MqttClient client;
	
	AtomicBoolean isConnLost;
	
	public MosquittoClientMqttCallback(SocketNode node,MqttClient client,SocketCallback callback){
		this.node = node;
		this.callback = callback;
		this.client = client;
		this.isConnLost = new AtomicBoolean(false);
	}

	@Override
	public void connectionLost(Throwable cause) {
		if(isConnLost.compareAndSet(false, true)) {
			final SocketEvent event = new SocketEvent(node,cause);
			node.runTask(new Runnable() {
				@Override
				public void run() {
					callback.connectionLost(event);
				}
			});
			if(StringUtils.isNotBlank(node.getConfig().getTestTopic())){
				new TestConnectThread().start();
			}
		}
	}

	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception {
		MqttSocketMessage msg = new MqttSocketMessage(topic,message);
		final SocketEvent event = new SocketEvent(node,msg);
		event.getSocketNode().setLastReceiveDate(new Date());
		node.runTask(new Runnable() {
			@Override
			public void run() {
				callback.messageArrived(event);
			}
		});
	}

	@Override
	public void deliveryComplete(IMqttDeliveryToken token) {
		try {
			token.waitForCompletion();
		} catch (MqttException e) {
			log.error(e.getMessage(),e);
		}
		final SocketEvent event = new SocketEvent(node);
		event.getSocketNode().setLastSendDate(new Date());
		node.runTask(new Runnable() {
			@Override
			public void run() {
				callback.deliveryComplete(event);
			}
		});
		if(isConnLost.compareAndSet(true, false)) {
			node.runTask(new Runnable() {
				@Override
				public void run() {
					callback.connected(event);
				}
			});
		}
	}
	
	private class TestConnectThread extends Thread{

		private Date date;
		private Integer errorCount = 0;

		public TestConnectThread(){
			this.date = new Date();
		}
		
		public void run() {
			log.info("测试连接线程启动");
			long timeout = node.getConfig().getKeepAliveInterval() * 4;
			long sleepTimes = node.getConfig().getKeepAliveInterval() * 1000;
			while (isConnLost.get()) {
				try {
					SocketMessage message = new MqttSocketMessage(node.getConfig().getTestTopic(), 1, "hello".getBytes());
					node.writeAndFlush(message);
					Thread.sleep(sleepTimes);
				} catch (Exception e) {
					log.error(e.getMessage(), e);
					errorCount++;
				}
				long times = DateTools.differenceSecond(new Date(), date);

				if (errorCount > 3 || times > timeout) {
					log.error("测试连接触发重启，错误次数：{{0}}，耗时：{{1}}秒，超时阀值：{{2}}秒", errorCount, times, timeout);
					isConnLost.compareAndSet(true, false);
					//触发重新启动
					node.reconnect();
				}
			}
			log.info("测试连接线程关闭");
		}
	}
}
